package com.hexinfo.dmpro.sparing.async;


import cn.hutool.core.date.BetweenFormater;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.io.file.FileAppender;
import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.StrUtil;
import com.hexinfo.dmpro.common.utils.ScanCommonConstants;
import com.hexinfo.dmpro.sparing.dto.CreateTableDTO;
import com.hexinfo.dmpro.sparing.dto.IgnoreListDTO;
import com.hexinfo.dmpro.sparing.dto.ScanSyncRuleDTO;
import com.hexinfo.dmpro.sparing.dto.SyncTypeDTO;
import com.hexinfo.dmpro.sparing.model.ScanMetadata;
import com.hexinfo.dmpro.sparing.model.SyncCompleteLibrary;
import com.hexinfo.dmpro.sparing.model.SyncCompleteLibraryTask;
import com.hexinfo.dmpro.sparing.service.*;
import com.hexinfo.dmpro.sparing.util.Logger;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.AsyncResult;
import org.springframework.stereotype.Component;

import java.util.*;
import java.util.concurrent.Future;

@Component
@AllArgsConstructor
@Slf4j
public class AsyncTask {

    private final ScanMetadataService scanMetadataService;

    private final SyncCompleteLibraryService syncCompleteLibraryService;

    private final ImpalaService impalaService;

    /**
     * 批量修改同步表信息
     * @param syncRuleDTO
     * @return
     * @throws Exception
     */
    @Async
    public Future<Boolean> updateToSyncType(ScanSyncRuleDTO syncRuleDTO) throws Exception {
        boolean upb = false;
        //查询 过滤掉已确认
        List<ScanMetadata> scanMetadataList = scanMetadataService.qyScanMetadataWhereYQR(syncRuleDTO);
        //添加同步规则ID和修改同步状态
        if(scanMetadataList!=null&&scanMetadataList.size()>0){
            scanMetadataList.stream().forEach(scan ->{
                scan.setSyncType(ScanCommonConstants.ClusterName.YQR.value);
                scan.setSyncRuleId(syncRuleDTO.getId());
            });
            upb = scanMetadataService.updateBatchById(scanMetadataList,10000);

        }
        if(upb){
            log.warn("=========已修改全量同步表信息"+scanMetadataList.size()+"条");
        }
        return new AsyncResult<>(true);
    }

    /**
     * 批量修改同步表信息
     * @param metadataList
     * @return
     * @throws Exception
     */
    @Async
    public Future<Boolean> updateBatchScanMetadata(List<ScanMetadata> metadataList) throws Exception {
        //添加同步规则ID和修改同步状态
        if(metadataList!=null&&metadataList.size()>0){
            boolean upb = scanMetadataService.updateBatchById(metadataList,10000);
            if(upb){
                log.warn("=========已修改全量同步表信息"+metadataList.size()+"条");
            }
        }
        return new AsyncResult<>(true);
    }

    /**
     *
     * @param dataRefreshService
     * @param task 任务信息
     * @param tableName 表名称
     * @param prefix 库名前缀
     * @param auditSql 稽核sql
     * @param ignoreListDTO 忽略名单
     * @return
     */
    @Async("threadPoolTaskExecutor")
    public Future<Map<String,Object>> asyncDatabase(DataRefreshService dataRefreshService, SyncCompleteLibraryTask task, String tableName, String prefix, String auditSql, IgnoreListDTO ignoreListDTO){
        Map<String,Object> result=new HashMap<String,Object>();
        log.warn(Thread.currentThread().getName());
        //当前开始时间
        Date staDateSx = DateUtil.date();
        SyncCompleteLibrary librarySx = new SyncCompleteLibrary();
        librarySx.setTaskName(task.getTaskName());
        librarySx.setTaskId(task.getId());
        librarySx.setLibraryTableName(tableName);
        librarySx.setLog("["+DateUtil.now()+"]【表刷新开始】");
        librarySx.setTaskState(ScanCommonConstants.ClusterName.JXZ.value);
        syncCompleteLibraryService.saveOrUpdate(librarySx);
        //表刷新
        String oa = dataRefreshService.souTableIsFinishDatabase(prefix,librarySx.getLibraryTableName());
//        oa = "成功";
        if(oa.contains("异常")){
            librarySx.setLog(librarySx.getLog()+"\n["+DateUtil.now()+"]【表刷新失败】"+oa);
            task.setTaskState(ScanCommonConstants.ClusterName.SB.value);
            librarySx.setTaskState(ScanCommonConstants.ClusterName.SB.value);
            syncCompleteLibraryService.saveOrUpdate(librarySx);
        }else{
            librarySx.setLog(librarySx.getLog()+"\n"+oa);
            syncCompleteLibraryService.saveOrUpdate(librarySx);
        }
        String auditSqlJQ = StrUtil.format(auditSql,tableName);
        String auditSqlWG = StrUtil.format(auditSql,prefix + tableName);
        librarySx.setLog(librarySx.getLog()+"\n["+DateUtil.now()+"]【表稽核开始】");
        syncCompleteLibraryService.saveOrUpdate(librarySx);
        //重试机制
        for (int i = 0; i < 4; i++) {
            boolean bol  = tableAudit(tableName,auditSqlJQ,auditSqlWG,librarySx,task,ignoreListDTO,i);
            if(bol){
                break;
            }
        }
        //当前开始时间
        Date endDateSx = DateUtil.date();
        librarySx.setHotStandbyTime(DateUtil.formatBetween(staDateSx,endDateSx, BetweenFormater.Level.SECOND));
        boolean bol = syncCompleteLibraryService.saveOrUpdate(librarySx);
        result.put(prefix+tableName,bol);
        return new AsyncResult<>(result);
    }

    /**
     * 表稽核
     */
    public boolean tableAudit(String tableName,String auditSqlJQ, String auditSqlWG, SyncCompleteLibrary librarySx,SyncCompleteLibraryTask task,IgnoreListDTO ignoreListDTO,int i){
        //表稽核
        boolean bol = true;
        Map<String,Object> mapJQ = new HashMap<>();
        Map<String,Object> mapWG = new HashMap<>();
        librarySx.setLog(librarySx.getLog()+"\n["+DateUtil.now()+"]【Impala】"+"\n["+DateUtil.now()+"]【金桥稽核SQL】"+auditSqlJQ+"\n["+DateUtil.now()+"]【万国稽核SQL】"+auditSqlWG);
        mapJQ = impalaService.countJQ(auditSqlJQ);
        mapWG = impalaService.count(auditSqlWG);
        String auditMessageJQ = (String)mapJQ.get("error");
        String auditMessageWG = (String)mapWG.get("error");
        if(StrUtil.isNotBlank(auditMessageJQ)||StrUtil.isNotBlank(auditMessageWG)){
            if(StrUtil.isBlank(auditMessageJQ)){
                Long countJQ  = Long.valueOf(String.valueOf(mapJQ.get("count")));
                auditMessageJQ = "金桥数据量为【"+countJQ+"】";
                librarySx.setSourceCountNumber(countJQ);
            }
            if(StrUtil.isBlank(auditMessageWG)){
                Long countWG  = Long.valueOf(String.valueOf(mapWG.get("count")));
                auditMessageWG = "万国数据量为【"+countWG+"】";
                librarySx.setSourceCountNumber(countWG);
            }
            if(i<3){
                librarySx.setLog(librarySx.getLog()+"\n["+DateUtil.now()+"]"+auditMessageJQ+"\n["+DateUtil.now()+"]"+auditMessageWG+"\n【重新稽核】");
            }else{
                librarySx.setLog(librarySx.getLog()+"\n["+DateUtil.now()+"]"+auditMessageJQ+"\n["+DateUtil.now()+"]"+auditMessageWG);
                librarySx.setTaskState(ScanCommonConstants.ClusterName.SB.value);
                task.setTaskState(ScanCommonConstants.ClusterName.SB.value);
            }
            syncCompleteLibraryService.saveOrUpdate(librarySx);
            bol = false;
        }else{
            Long countJQ  = Long.valueOf(String.valueOf(mapJQ.get("count")));
            Long countWG  = Long.valueOf(String.valueOf(mapWG.get("count")));
            librarySx.setLog(librarySx.getLog()+"\n["+DateUtil.now()+"]"+"金桥数据量为【"+countJQ+"】\n["+DateUtil.now()+"]万国数据量为【"+countWG+"】");
            librarySx.setSourceCountNumber(countJQ);
            librarySx.setTargetCountNumber(countWG);
            if(countJQ.equals(countWG)){
                librarySx.setTaskState(ScanCommonConstants.ClusterName.YWC.value);
                librarySx.setLog(librarySx.getLog()+"\n["+DateUtil.now()+"]【对比成功】\n["+DateUtil.now()+"]【表稽核结束】");
                syncCompleteLibraryService.saveOrUpdate(librarySx);
            }else{
                //模糊匹配
                String tblNameFlag = "常规";
                if(ObjectUtil.isNotEmpty(ignoreListDTO.getHmList())){
                    for (String ign: ignoreListDTO.getHmList()) {
                        //稽核忽略
                        if(tableName.contains(ign)){
                            log.warn("【模糊忽略表】"+tableName+"【规则】"+ign);
                            tblNameFlag = ScanCommonConstants.ClusterName.MH.value;
                            librarySx.setTaskState(ScanCommonConstants.ClusterName.HL.value);
                            break;
                        }
                    }
                }
                //精准匹配
                if("常规".equals(tblNameFlag)&&ObjectUtil.isNotEmpty(ignoreListDTO.getJzList())){
                    for (String ign: ignoreListDTO.getJzList()) {
                        //稽核忽略
                        if(tableName.equals(task.getDatabaseName()+"."+ign)){
                            Logger.log("【精准忽略表】"+tableName+"【规则】"+task.getDatabaseName()+"."+ign);
                            librarySx.setTaskState(ScanCommonConstants.ClusterName.HL.value);
                            break;
                        }
                    }
                }
                if(!ScanCommonConstants.ClusterName.HL.value.equals(librarySx.getTaskState())){
                    task.setTaskState(ScanCommonConstants.ClusterName.SB.value);
                    librarySx.setLog(librarySx.getLog()+"\n["+DateUtil.now()+"]【对比失败】\n["+DateUtil.now()+"]【表稽核结束】");
                    librarySx.setTaskState(ScanCommonConstants.ClusterName.SB.value);
                }
                syncCompleteLibraryService.saveOrUpdate(librarySx);
            }
        }
        return bol;
    }


    /**
     * DISTCP方法
     */
    @Async("threadPoolTaskExecutor")
    public Future<Map<String,Object>> execShell(String hdfsP, String heatBackup,FileAppender coldLog) {
        Map<String,Object> result=new HashMap<String,Object>();
        Date dateSta = DateUtil.date();
        String shell = heatBackup + " " + hdfsP;
        logLog(coldLog,"【开始同步】"+shell,"1");
        coldLog.flush();
        int i = execShellCodeDISTCP(shell,coldLog);
        if (i != 0){
            logLog(coldLog,"DistCP命令：" + shell+"  --执行结果："+i+"（失败）","1");
        }else {
            Date dateEnd = DateUtil.date();
            logLog(coldLog,"DistCP命令：" + shell+"  --执行结果："+i+"（成功）","3");
            logLog(coldLog,"【同步用时】"+hdfsP+"【开始时间】"+DateUtil.formatDateTime(dateSta)+"【结束时间】"+DateUtil.formatDateTime(dateEnd),"3");
        }
        result.put(hdfsP,i);
        return new AsyncResult<>(result);
    }

    /**
     * 执行shell脚本，只返回执行结果
     * @param pathOrCommand 脚本路径或者命令
     * @return
     */
    private int execShellCodeDISTCP(String pathOrCommand, FileAppender coldLog) {
        int exitValue = 0;
        try {
            Process ps = Runtime.getRuntime().exec(pathOrCommand);
            exitValue = ps.waitFor();
            //重试2次
            if (exitValue != 0){
                for (int j = 1; j<3; j++){
                    Process ps1 = Runtime.getRuntime().exec(pathOrCommand);
                    exitValue = ps1.waitFor();
                    if(exitValue == 0){
                        break;
                    }
                }
            }
        } catch (Exception e) {
            log.error("执行DistCP脚本失败命令：{}，执行DistCP脚本失败报错：{}",pathOrCommand,e);
            logLog(coldLog,"【执行DistCP错误】shell："+pathOrCommand+"---执行异常："+e,"1");
            return 999;
        }
        return exitValue;
    }

    /**
     * 加锁，防止并发问题
     * @param tableLog
     * @param logStr
     * @param str
     */
    private synchronized void logLog(FileAppender tableLog, String logStr ,String str) {
        if (str.equals("1")){
            tableLog.append(logTime(logStr));
        }else if (str.equals("2")){
        }else if (str.equals("3")){
            tableLog.append(logTime(logStr));
        }else if (str.equals("4")){
            tableLog.flush();     //表日志刷进日志中
        }
    }

    /**
     * 创表
     * @param dto
     * @param library
     */
    @Async("threadPoolTaskExecutor")
    public Future<Map<String,Object>> forCreateTable(CreateTableDTO dto,SyncCompleteLibrary library,Map<String,Integer> map){
        Map<String,Object> result=new HashMap<String,Object>();
        Map<String,Object> mapJQ = impalaService.showTableHive(dto.getDbName()+"."+dto.getTableName());
        String showJQ = (String)mapJQ.get("error");
        String createSQLWG = "";
        if(StrUtil.isNotBlank(showJQ)){
            library.setLog(library.getLog()+"\n["+DateUtil.now()+"]【查询建表SQL异常】"+showJQ);
            map.put("SB",map.get("SB")+1);
            result.put(dto.getDbName()+"."+dto.getTableName(),createSQLWG);
        }else{
            String createWG = (String)mapJQ.get("show");
//            //创表语句截取
//            String[] creates = createWG.split("LOCATION");
//            if(ObjectUtil.isNotEmpty(creates)){
//                if(creates.length>1){
//                    createSQLWG = creates[0];
                    createSQLWG = createWG.replace("CREATE TABLE","CREATE TABLE IF NOT EXISTS")
                                            .replace("CREATE VIEW","CREATE VIEW IF NOT EXISTS")
                                            .replace("CREATE EXTERNAL TABLE","CREATE EXTERNAL TABLE IF NOT EXISTS");
                    Map<String,Object> mapWG = impalaService.createTableHive(createSQLWG,dto.getDbName()+"."+dto.getTableName());
                    String sql = (String)mapWG.get("sql");
                    library.setLog(library.getLog()+"\n["+DateUtil.now()+"]【建表SQL】\n"+sql);
                    String errprWG = (String)mapWG.get("error");
                    if(StrUtil.isNotBlank(errprWG)){
                        library.setLog(library.getLog()+"\n["+DateUtil.now()+"]【建表SQL异常】"+dto.getDbName()+"."+dto.getTableName()+"\n"+errprWG);
                        map.put("SB",map.get("SB")+1);
                        result.put(dto.getDbName()+"."+dto.getTableName(),ScanCommonConstants.ClusterName.SB.value);
                    }else{
                        library.setLog(library.getLog()+"\n["+DateUtil.now()+"]【建表成功】"+dto.getDbName()+"."+dto.getTableName());
                        map.put("YWC",map.get("YWC")+1);
                        result.put(dto.getDbName()+"."+dto.getTableName(),ScanCommonConstants.ClusterName.YWC.value);
                    }
//                }
//            }
        }
        syncCompleteLibraryService.saveOrUpdate(library);
        return new AsyncResult<>(result);
    }

    /**
     * 创表
     */
    @Async("threadPoolTaskExecutor")
    public Future<Map<String,Object>> forCreateTable(CreateTableDTO dto,FileAppender heatlog,Map<String,Integer> map){
        Map<String,Object> result=new HashMap<String,Object>();
        Map<String,Object> mapJQ = impalaService.showTableHive(dto.getDbName()+"."+dto.getTableName());
        String showJQ = (String)mapJQ.get("error");
        String createSQLWG = "";
        if(StrUtil.isNotBlank(showJQ)){
            heatlog.append(logTime("【查询建表SQL异常】"+showJQ));
            map.put("SB",map.get("SB")+1);
            result.put(dto.getDbName()+"."+dto.getTableName(),createSQLWG);
        }else{
            String createWG = (String)mapJQ.get("show");
            createSQLWG = createWG.replace("CREATE TABLE","CREATE TABLE IF NOT EXISTS")
                            .replace("CREATE VIEW","CREATE VIEW IF NOT EXISTS")
                            .replace("CREATE EXTERNAL TABLE","CREATE EXTERNAL TABLE IF NOT EXISTS");
            Map<String,Object> mapWG = impalaService.createTableHive(createSQLWG,dto.getDbName()+"."+dto.getTableName());
            String sql = (String)mapWG.get("sql");
            heatlog.append(logTime("【建表SQL】"+sql));
            String errprWG = (String)mapWG.get("error");
            if(StrUtil.isNotBlank(errprWG)){
                heatlog.append(logTime("【建表SQL异常】"+dto.getDbName()+"."+dto.getTableName()+"\n"+errprWG));
                map.put("SB",map.get("SB")+1);
                result.put(dto.getDbName()+"."+dto.getTableName(),ScanCommonConstants.ClusterName.SB.value);
            }else{
                heatlog.append(logTime("【建表成功】"+dto.getDbName()+"."+dto.getTableName()));
                map.put("YWC",map.get("YWC")+1);
                result.put(dto.getDbName()+"."+dto.getTableName(),ScanCommonConstants.ClusterName.YWC.value);
            }
        }
        //文件信息落地
        heatlog.flush();
        return new AsyncResult<>(result);
    }

    /**
     * 日志加日期时间
     * @param str
     * @return
     */
    private String logTime(String str) {
        return "["+ DateUtil.format(DateUtil.date(), "yyyy-MM-dd HH:mm:ss")+"]" +str;
    }
}
